import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Time;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;

import java.io.IOException;
import java.util.List;

public class ElasticsearchPITExample {

    public static void main(String[] args) throws IOException {
        // 创建凭证提供者
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials("elastic", "6GaCyvL7yNyvuysbFLuE"));

        // 构建 RestClient
        RestClient restClient = RestClient.builder(new HttpHost("192.168.6.217", 9200, "http"))
                .setHttpClientConfigCallback(httpClientBuilder ->
                        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider))
                .build();

        ElasticsearchTransport transport = new RestClientTransport(
                restClient,
                new JacksonJsonpMapper() // JSON处理器
        );

        ElasticsearchClient client = new ElasticsearchClient(transport);

        try {
            // 2. 创建PIT（Point-In-Time）
            String pitId = createPIT(client, "ctms_forecast_log_2022_test");

            // 3. 使用PIT查询
            searchWithPIT(client, pitId);

            // 4. 关闭PIT（必须显式释放资源）
            closePIT(client, pitId);

        } finally {
            restClient.close(); // 关闭连接
        }
    }

    /**
     * 创建PIT
     */
    private static String createPIT(ElasticsearchClient client, String indexName) throws IOException {
        OpenPointInTimeResponse response = client.openPointInTime(b -> b
                .index(indexName)
                .keepAlive(Time.of(t -> t.time("1m"))) // PIT有效期1分钟
        );

        String pitId = response.id();
        System.out.println("Created PIT with ID: " + pitId);
        return pitId;
    }

    /**
     * 使用PIT查询
     */
    private static void searchWithPIT(ElasticsearchClient client, String pitId) throws IOException {
        SearchResponse<String> response = client
                .search(s -> s.pit(p -> p.id(pitId).keepAlive(Time.of(t -> t.time("15m")))) // 关联PIT并续期
                                .query(q -> q.matchAll(m -> m))                // 查询所有文档
                                .size(10)                                     // 返回10条结果
                        , String.class                                   // 返回结果类型
                );

        // 处理结果
        List<Hit<String>> hits = response.hits().hits();
        for (Hit<String> hit : hits) {
            System.out.println("Found document: " + hit.source());
        }
    }

    /**
     * 关闭PIT
     */
    private static void closePIT(ElasticsearchClient client, String pitId) throws IOException {
        client.closePointInTime(c -> c.id(pitId));
        System.out.println("Closed PIT: " + pitId);
    }
}